package cn.doitedu.demo.connectors;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class MysqlCdc连接器使用 {
    public static void main(String[] args) {

        // 构造stream api 编程的环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);

        // 构造 sql 编程环境
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 创建一个mysql-cdc连接器表

        tenv.executeSql("CREATE TABLE score_mysql (\n" +
                "     id INT,                     \n" +
                "     name string,                \n" +
                "     gender STRING,              \n" +
                "     score double,               \n" +
                "     PRIMARY KEY(id) NOT ENFORCED\n" +
                "     ) WITH (                   \n" +
                "     'connector' = 'mysql-cdc', \n" +
                "     'hostname' = 'doitedu',    \n" +
                "     'port' = '3306',           \n" +
                "     'username' = 'root',       \n" +
                "     'password' = 'root',       \n" +
                "     'database-name' = 'flinktest',\n" +
                "     'table-name' = 'flink_score')");

        tenv.executeSql("select * from score_mysql").print();
    }
}
